Parquet v3 read concurrency - #2235
Conversation
Budget memory and threads separately per read stage instead of one shared fraction, and add a ColumnDataPrefetch stage that issues the compressed data-page reads (charged to its own memory budget) while ColumnData only decodes. The old single 0.2 fraction capped the data stage at 0.2 of both memory and threads, so only ~2 row groups were read/decoded ahead and the S3 link sat idle on latency-bound, high-RTT reads. Now compressed reads run deep (cheap per row group) while decoded row groups stay bounded, hiding per-GET latency. Also reconcile the decoded-memory charge up to the actual footprint inside decodePrimitiveColumn, before formOutputColumn moves the column, so the honest cap actually bounds decode-ahead. Squashed extraction of 34816a3 + 114640e + f260506 from the parquet-v3 feature branch onto antalya-26.6. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
ad5f3aa to
79d3658
Compare
|
Could you add Changelog entry please? |
mkmkme
left a comment
mkmkme
left a comment
There was a problem hiding this comment.
Looks good from the brief look
|
Will do (and benchmark justification), but i would ask to start reviewing of it tomorrow, i want to do some small changes and remove some slop. :) |
| /// Budget memory and threads separately: a single 0.2 fraction capped ColumnData at 0.2 of both, | ||
| /// so only ~2 row groups were read/decoded ahead. Give ColumnData most of the memory and threads | ||
| /// (deep, decode-bound); give the small latency-bound index/bloom reads threads for parallelism. | ||
| using S = ReadStage; |
There was a problem hiding this comment.
Useless alias. Keeps several characters in each line, but adds cognitive complexity.
| set_fractions(S::ColumnDataPrefetch, 0.45, 1); | ||
| /// Decode: bounded memory (decoded row groups are large) but most threads. Caps resident decoded | ||
| /// row groups independently of prefetch depth. | ||
| set_fractions(S::ColumnData, 0.30, 3); |
There was a problem hiding this comment.
Each stage get part of memory/thread budget based on some magic numbers.
I trust that numbers has a ground, but good to add it somewhere.
ca7a4f6 to
79d3658
Compare
The reader scheduler splits a static memory and thread budget across the read stages (bloom/index reads, compressed prefetch, decode) using fixed relative weights. Those weights are the main lever for latency-bound object-storage reads - the compressed prefetch weight sets the read-ahead depth, the decode thread weight sets decode concurrency - but they were hardcoded, so tuning them required a rebuild. Add `input_format_parquet_read_stage_weights`, an experimental `Map` setting. Keys are `<stage>.<resource>` (resource is `memory` or `threads`); values are relative weights, normalized per resource across stages exactly like the built-in defaults. Only the listed keys override; unspecified stages keep their defaults, so an empty map (the default) reproduces the previous behavior bit for bit. Unknown stage or resource names, negative weights, and a configuration that zeroes out the whole memory or thread budget all raise `BAD_ARGUMENTS` rather than silently falling back. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: UnamedRus <dtitmoav@gmail.com>
…atches `ColumnDataPrefetch` tasks do all their work (`determinePagesToPrefetch` + `startPrefetch`) synchronously in `scheduleTask`; their `runTask` body is empty and the reads proceed asynchronously in the Prefetcher's io pool. Their run time is therefore ~0, but `cost_estimate_bytes` was set to the charged compressed bytes, which are large. `scheduleTasksIfNeeded` uses `cost_estimate_bytes` as a proxy for run time to group tiny tasks into batches. Reporting the large compressed-byte charge made the batcher split these zero-work tasks across up to `parsing_threads` batches - one thread-pool dispatch each - for tasks that do nothing on the thread. Report a cost of 0 for `ColumnDataPrefetch` tasks so they collapse into a single batch, removing the spurious per-subgroup dispatch overhead introduced by the prefetch/decode stage split. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
ccd3c2d to
df7105e
Compare
Condense the oversized comment blocks added by the read-concurrency work (the stage-weight rationale in `ReadManager::init`, the ColumnDataPrefetch/ColumnData scheduleTask and runTask notes, and the cost-estimate exception) to a few lines each. No code changes. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
| format_settings.parquet.enable_json_parsing = settings[Setting::input_format_parquet_enable_json_parsing]; | ||
| format_settings.parquet.memory_low_watermark = settings[Setting::input_format_parquet_memory_low_watermark]; | ||
| format_settings.parquet.memory_high_watermark = settings[Setting::input_format_parquet_memory_high_watermark]; | ||
| { |
There was a problem hiding this comment.
Parsing like this already exists for http_response_headers setting. Suggestion is to move login into template function and reuse for both settings.
https://github.com/Altinity/ClickHouse/blob/antalya-26.6/src/Interpreters/executeQuery.cpp#L2447
| Approximate memory limit for Parquet reader v3. Limits how many row groups or columns can be read in parallel. When reading multiple files in one query, the limit is on total memory usage across those files. | ||
| )", 0) \ | ||
| DECLARE(Map, input_format_parquet_read_stage_weights, "", R"( | ||
| Expert tuning knob for the Parquet reader scheduler. Overrides the relative memory and thread budget weights assigned to each read stage. Keys have the form `<stage>.<resource>`, where `<resource>` is `memory` or `threads` and `<stage>` is one of `bloom_filter_header`, `bloom_filter_blocks_or_dictionary`, `column_index_and_offset_index`, `offset_index`, `column_data_prefetch`, `column_data`. Values are relative weights (they need not sum to 1; each is normalized against the sum of that resource across stages). Only the specified keys override the built-in defaults; unspecified stages keep their defaults. Example: `{'column_data_prefetch.memory': 12, 'column_data.threads': 4}`. Advanced and experimental: the internal stage model may change between versions. |
There was a problem hiding this comment.
unspecified stages keep their defaults
How user can know default values except reading source code? Now if I change some value, I even have no idea it increased or decreased.
And key-value pairs in string setting when keys must be in short predefined list is weird, not as all other settings in CH.
Only one sample like this is http_response_headers, but http_response_headers has unspecified list of possible keys.
12 different settings instead look just as good as this.
There was a problem hiding this comment.
Decided to replace with 2 float setting, which actually encode 90% of useful stuff (ie between prefetch and decode)
input_format_parquet_prefetch_memory_fraction
input_format_parquet_decode_thread_fraction
…nobs
The `input_format_parquet_read_stage_weights` `Map` setting failed to build:
`Map` is not in `FORMAT_FACTORY_SETTINGS_SUPPORTED_TYPES` (nor in the supported
types of the collections that embed the format settings, e.g.
`DataLakeStorageSettings`), so the settings-layout machinery has no
`SettingTypeTag::Map` for those collections:
DataLakeStorageSettings.cpp: error: no member named 'Map' in
'DataLakeStorageSettingsTraitsDetail::SettingTypeTag'
...expanded from DECLARE(Map, input_format_parquet_read_stage_weights, ...)
Adding `Map` to every such collection's supported-types list is invasive and is
why no format setting is a `Map` today. Replace the single `Map` with two `Float`
settings (`Float` is already a supported format-settings type):
- `input_format_parquet_prefetch_memory_fraction` (default 0.6): of the memory
budget reserved for column data, the share given to compressed read-ahead
(`ColumnDataPrefetch`) vs decoded output (`ColumnData`).
- `input_format_parquet_decode_thread_fraction` (default 0.375): `ColumnData`'s
share of the parsing thread pool; the read-issuing stages split the rest.
These cover the two binding constraints on decode-ahead depth (memory on
latency-bound remote reads, threads on CPU-bound local decode) without exposing
the dead per-stage knobs for the read-issuing stages. `ReadManager::init` now
computes per-stage fractions directly from the two knobs; the defaults reproduce
the previously hard-coded fractions exactly, so behavior is unchanged by default.
The index/bloom stages keep fixed shares.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The native reader is now the only one (input_format_parquet_use_native_reader_v3 is obsolete/always-on), so the "v3" qualifier in the user-facing setting descriptions and code comments is stale. Identifiers and the obsolete-setting history entries keep their names. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The two knobs were stored both in FormatSettings.parquet and copied into SharedResourcesExt (mirroring the memory watermarks). The watermarks live in the ext because getLimitsPerReader (a static method without options access) reads them; the fractions are only consumed in ReadManager::init, which already has reader.options.format.parquet. Drop the duplicate ext fields and the copy, read the fractions directly, and spell the locals out in full. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
|
|
Failed to start ClickHouse server due to a DNS error:
|
Ran locally |
CI triage for #2235 (head
|
5 participants
Number of changes to bring parquet v3 reader perf closer to arrow based
Changelog category (leave one):
Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):